package com.jsu.server;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.jsu.sender.MessageSender;
import com.jsu.service.BarrageService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import com.jsu.client.UserClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocketServer {

    private static BarrageService barrageService;
    @Autowired(required = false)
    private UserClient userClient;

    @Autowired
    public void setUserService (BarrageService barrageService){
        WebSocketServer.barrageService = barrageService;
    }

    //记录当前在线数量
    private static int onLineCount = 0;
    //记录总共发送弹幕条数
    private static int barrageCount = 0;
    //每个客户端和对应的websocket连接
    public static ConcurrentHashMap<String,WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    //与客户端的连接对话，通过session向客户端发送信息
    private Session session;
    //userId
    private String userId = "";


    /**
     * 连接建立成功调用方法
     * @param session 对话
     * @param userId 用户id
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId){
        this.session = session;
        this.userId = userId;
        //相当于刷新重进
        if (webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);
            webSocketMap.put(userId,this);
        }
        else {
            webSocketMap.put(userId,this);
            addOnlineCount();//在线人数加一
        }
        log.info("用户"+userId+"已连接"+"，当前在线人数："+getOnLineCount());
        try {
            sendMessage("连接成功！");
        } catch (IOException e) {
            log.error("用户:"+userId+",网络异常!!!!!!");
        }
    }

    /**
     * 用户连接关闭以后方法调用
     * @param session
     * @param userId
     */
    @OnClose
    public void onClose(Session session,@PathParam("userId") String userId){
        if(webSocketMap.containsKey(userId)){
            webSocketMap.remove(userId);//移出map，减掉在线人数
            subOnlineCount();
        }
        log.info("用户"+userId+"已退出"+"，当前在线人数："+getOnLineCount());
    }

    /**
     * 收到客户端消息后调用
     * @param session
     * @param message 客户端传来的信息
     */
    @OnMessage
    public void onMessage(Session session,String message){
        log.info("用户消息："+userId+"，报文："+message);
        //消息保存到数据库、redis，可以进行群发
        if(StringUtils.isNotBlank(message)){
            JSONObject jsonObject = JSON.parseObject(message);//解析报文
            Integer sid = Integer.valueOf(jsonObject.getString("sid"));
            Integer type = Integer.valueOf(jsonObject.getString("type"));
            //String agendaId = "1";
            //向数据库存储弹幕信息
            barrageService.addBarrage(sid,type,message, String.valueOf(userId));
            //TODO  直播禁言
            Integer liveState = (Integer) barrageService.getUserLiveBan(Integer.valueOf(userId),sid,type).getData();
            if (liveState == 1){
                sendAllMessage(message);//将信息发给所有用户进行查看
            }
            else {
                JSONObject error = new JSONObject();
                error.put("type", "error");
                error.put("message", "已被禁言，无法发送！");
                try {
                    session.getBasicRemote().sendText(error.toJSONString());
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
        }
        else {
            JSONObject error = new JSONObject();
            error.put("type", "error");
            error.put("message", "弹幕发送失败！");
            try {
                session.getBasicRemote().sendText(error.toJSONString());
            } catch (IOException e) {
                throw new RuntimeException(e);
            }

        }
    }

    @OnError
    public void onError(Session session,Throwable error){
        log.error("用户"+this.userId+"发生错误，原因："+error);
        error.printStackTrace();
    }

    /**
     * 实现服务器主动向客户端发送消息
     */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }

    /**
     * 发送自定义信息，单点消息,直播订阅通知
     */
    public static void sendInfo(String message,@PathParam("userId") String userId){

        log.info("发送消息到："+userId+"，消息内容是："+message);
        if(StringUtils.isNotBlank(userId)&&webSocketMap.containsKey(userId)){
            try {
                webSocketMap.get(userId).session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                throw new RuntimeException("发送失败！");
            }
        }
        else {
            log.error("用户"+userId+"不在线！");
        }
    }
    //
    public static void sendAllMessage(String message){
        log.info("消息发送到全体："+message);
        barrageCount++;
        String formatBarrageMessage = formatBarrageMessage(message);
        String formatPopularityMessage = formatPopularityMessage(calPopularity());
        //遍历用户发送
        for (Map.Entry<String, WebSocketServer> webSocketServerEntry : webSocketMap.entrySet()) {
            //String key = webSocketServerEntry.getKey();
            try {

                webSocketServerEntry.getValue().session.getBasicRemote().sendText(formatBarrageMessage);
                webSocketServerEntry.getValue().session.getBasicRemote().sendText(formatPopularityMessage);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }



    // 用来判断连接是否是管理员连接
    public static boolean isAdmin(Session session) {
        return session.getPathParameters().get("userId").equals("admin");
    }

    // 然后，在需要向管理员发送消息的地方调用这个方法来判断
    public static void sendToAdminOnlineUsers() {
        JSONArray onlineUsers = new JSONArray();
        //拿到在线成员userId
        onlineUsers.addAll(webSocketMap.keySet());

        JSONObject message = new JSONObject();
        message.put("type", "onlineUsers");
        message.put("message", onlineUsers);

        for (Map.Entry<String, WebSocketServer> entry : webSocketMap.entrySet()) {
            WebSocketServer server = entry.getValue();
            if (isAdmin(server.session)) {
                try {
                    server.session.getBasicRemote().sendText(message.toJSONString());
                } catch (IOException e) {
                    log.error("发送消息给管理员失败: {}", e.getMessage());
                }
            }
            else {
                //发送延时队列
                MessageSender.sendDelayQueue(message.toJSONString(),1000 * 60 * 30);
            }
        }
    }

    //向管理员端推送文章审核
    public static void sendToAdminPassage(){
        JSONObject message = new JSONObject();
        message.put("type", "passage");
        message.put("message", "有文章需要进行审核，请前往审核系统进行审核");
        for (Map.Entry<String, WebSocketServer> entry : webSocketMap.entrySet()) {
            WebSocketServer server = entry.getValue();
            if (isAdmin(server.session)){
                try {
                    server.session.getBasicRemote().sendText(message.toJSONString());//推送提醒通知

                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
            //管理员不在线，存储到消息队列每隔三十分钟进行发送，ack手动确认后删除
            else{
                MessageSender.sendDelayQueue(message.toJSONString(),1000*60*30);
            }
        }
    }



    // 计算热度值
    private static Double calPopularity(){
        double popularity = 0.6 * onLineCount + 0.4 * barrageCount;
        return popularity;
    }


    // 封装弹幕消息
    public static String formatBarrageMessage(String content) {
        JSONObject message = new JSONObject();
        message.put("type", "barrage");
        message.put("content", content);
        return message.toJSONString();
    }

    // 封装热度值消息
    public static String formatPopularityMessage(double popularity) {
        JSONObject message = new JSONObject();
        String popularity2 = String.format("%.2f", popularity);
        message.put("type", "popularity");
        message.put("value", popularity2);
        return message.toJSONString();
    }

    //线程同步
    public static synchronized void addOnlineCount(){
        onLineCount++;
    }
    public static synchronized void subOnlineCount(){
        onLineCount--;
    }
    public static synchronized int getOnLineCount(){
        return onLineCount;
    }
}
